package com.example.kafka.kerberos.kafka;

import org.apache.kafka.clients.admin.AdminClient;
import org.apache.kafka.clients.admin.DescribeClusterResult;
import org.apache.kafka.clients.admin.KafkaAdminClient;
import org.apache.kafka.common.KafkaFuture;
import org.apache.kafka.common.Node;

import java.io.Closeable;
import java.io.IOException;
import java.util.Properties;

/**
 * @author stone
 * @version $1.0
 * @date 2020/9/15 15:12:44
 */
public class KafkaAdmins implements Closeable {

    private static final int DEFAULT_TIME_OUT_SECOND = 60;

    private static final String SASL_JAAS_CONFIG =
            "com.sun.security.auth.module.Krb5LoginModule required\n" +
            "useKeyTab=true\n" +
            "storeKey=true\n" +
            "keyTab=\"/home/stone/stone.keytab\"\n" +
            "principal=\"stone/stone.analyticservice.net@ANALYTICSERVICE.NET\";\n";

    private AdminClient adminClient;

    public KafkaAdmins(Properties adminProps) {
//        System.setProperty("java.security.krb5.conf", "/etc/krb5.conf");
//        System.setProperty("java.security.auth.login.config", "/home/stone/jaas.conf");
        // 配置jaas
        adminProps.put("sasl.jaas.config", SASL_JAAS_CONFIG);

        // 添加kerberos认证需要的参数
        adminProps.put("security.protocol", "SASL_PLAINTEXT");
        adminProps.put("sasl.mechanism", "GSSAPI");
        adminProps.put("sasl.kerberos.service.name", "kafka");
        this.adminClient = KafkaAdminClient.create(adminProps);
    }

    public AdminClient getAdminClient() {
        return adminClient;
    }

    @Override
    public void close() throws IOException {
        this.adminClient.close();
    }

    /**
     * 获取集群配置信息
     *
     * @return
     */
    public DescribeClusterResult describeCluster() {
        return this.adminClient.describeCluster();
    }

    public static void main(String[] args) throws IOException {
        Properties props = new Properties();
        props.put("bootstrap.servers", "kafka.analyticservice.net:9092");
        KafkaAdmins kafkaAdmins = new KafkaAdmins(props);
        AdminClient adminClient = kafkaAdmins.getAdminClient();
        DescribeClusterResult clusterResult = adminClient.describeCluster();
        KafkaFuture<Node> controller = clusterResult.controller();
        try {
            String host = controller.get().host();
            System.out.println(host);
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            kafkaAdmins.close();
        }

    }

}
